package com.example.rabbit.demora.config;

import com.example.rabbit.demora.mq.MqDemoLister;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.util.UUID;

/**
 * User: Amos
 * Date: 2019/8/2
 * Time: 15:12
 */
@Configuration
@Slf4j
public class RabbitMqConfig {
//    @Autowired
//    MqDemoLister mqDemoLister;
    @Bean
    public ConnectionFactory connectionFactory(@Value("${spring.rabbitmq.port}") int port,
                                               @Value("${spring.rabbitmq.host}") String host,
                                               @Value("${spring.rabbitmq.username}") String userName,
                                               @Value("${spring.rabbitmq.password}") String password,
                                               //消息发送确认，通过实现ConfirmCallBack接口，消息发送到交换器Exchange后触发回调。
                                               //spring.rabbitmq.publisher-confirms = true
                                               @Value("${spring.rabbitmq.publisher-confirms}") boolean isConfirm,
                                               //通过实现ReturnCallback接口，如果消息从交换器发送到对应队列失败时触发
                                               // （比如根据发送消息时指定的routingKey找不到队列时会触发）
                                               //spring.rabbitmq.publisher-returns = true----mandatory参数
                                               @Value("${spring.rabbitmq.publisher-returns}") boolean isRConfirm,
                                               @Value("${spring.rabbitmq.virtual-host}") String vhost){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setVirtualHost(vhost);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(userName);
        connectionFactory.setPassword(password);
        connectionFactory.setPublisherConfirms(isConfirm);
        connectionFactory.setPublisherReturns(isRConfirm);
        return connectionFactory;
    }


    @Bean
    /** 因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置 */
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate(@Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        //设置为true 根据路由key 找不到符合条件对列，返回给生产者，false 丢弃
        template.setMandatory(true);
        return template;
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("boot-test");
    }

    @Bean
    public Queue queue() {
        return new Queue("boot-test", true);
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with("boot-test");
    }



    /*************消费者配置************/
    /**
     *SimpleRabbitListenerContainerFactory
     * @param connectionFactory
     *
     * @return
     */
    @Bean
    SimpleMessageListenerContainer simpleContainer(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames("boot-test");
        //消费者并发配置
        container.setMaxConcurrentConsumers(5);
        container.setConcurrentConsumers(2);
        //消费者一次获取消息个数,如果处理不完会存在客户端
        //设置为BasicQos的值，当
        //prefetch_count在no_ask=false的情况下生效，即在自动应答的情况下这两个值是不生效的
        //会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，
        // 即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
        container.setPrefetchCount(2);
        //手动ack
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        /**
        //MessageHandler 定义一个普通类作为消费者类，指定方法，和消息转换器
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageHandler());
        //设置处理器的消费消息的默认方法,如果没有设置，那么默认的处理器中的默认方式是handleMessage方法
        adapter.setDefaultListenerMethod("onMessage");
        adapter.setMessageConverter(new Jackson2JsonMessageConverter());
        */

        //        container.setMessageListener(mqDemoLister);

        MessageListenerAdapter adapter = new MessageListenerAdapter(new MqDemoLister());
        adapter.setMessageConverter(new Jackson2JsonMessageConverter());
        container.setMessageListener(adapter);
        adapter.setDefaultListenerMethod("onMessage");


        //消费端的标签策略
        container.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String queue) {
                return queue + "_" + UUID.randomUUID().toString();
            }
        });

        //container.setMessageConverter(new Jackson2JsonMessageConverter());不建议使用
        return container;
    }
}
